package com.xiaowanzi.transporter.handler;

import com.xiaowanzi.transporter.constants.MessageTypeEnum;
import com.xiaowanzi.transporter.model.Message;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

public class ClientMessageHandler extends ServerMessageHandler {

  private ExecutorService executor = Executors.newSingleThreadExecutor();

  @Override
  public void channelActive(ChannelHandlerContext ctx) throws Exception {
    executor.execute(new MessageSender(ctx));
  }

  /**
   * 这里userEventTriggered()主要是在一些用户事件触发时被调用，这里我们定义的事件是进行心跳检测的ping和pong消息，
   * 当前触发器会在指定的触发器指定的时间返回内如果客户端没有被读取消息或者没有写入消息到管道，则会触发当前方法
   */
  @Override
  public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    if (evt instanceof IdleStateEvent) {
      IdleStateEvent event = (IdleStateEvent) evt;
      if (event.state() == IdleState.READER_IDLE) {
        Message message = new Message();
        message.setMessageType(MessageTypeEnum.PING);
        ctx.writeAndFlush(message);
      } else if (event.state() == IdleState.WRITER_IDLE) {
        ctx.close();
      }
    }
  }

  private static final class MessageSender implements Runnable {

    private static final AtomicLong counter = new AtomicLong(1);

    private volatile ChannelHandlerContext ctx;

    public MessageSender(ChannelHandlerContext ctx) {
      this.ctx = ctx;
    }

    @Override
    public void run() {
      try {
        while (true) {
          TimeUnit.SECONDS.sleep(new Random().nextInt(3));
          Message message = new Message();
          message.setMessageType(MessageTypeEnum.REQUEST);
          message.setBody("this is my " + counter.getAndIncrement() + " message.");
          message.addAttachment("name", "xufeng");
          ctx.writeAndFlush(message);
        }
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }
}
